package belf.migrate.api.sink.jdbc;

import belf.migrate.api.queue.RedisQueue;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;
import belf.migrate.api.table.schema.AbstractSinkCatalog;
import belf.migrate.api.taskconf.JobContext;
import belf.migrate.api.util.EngineConf;
import com.fasterxml.jackson.core.JsonProcessingException;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

@Slf4j
public class JdbcSinkWorker extends Thread {
    private static final long SLEEP_TIME = EngineConf.getInstance().getLong("jdbc.sink.interval", 3000);
    private volatile boolean paused = false;
    private volatile boolean terminated = false;

    private final JobContext jobContext;

    private final AbstractSinkCatalog sinkCatalog;

    private final int BATCH;

    private final JdbcBatchSaver jdbcBatchSaver;

    public JdbcSinkWorker(JobContext jobContext) {
        this.jobContext = jobContext;
        this.sinkCatalog = jobContext.getSinkCatalog();
        this.BATCH = jobContext.getTaskConf().getSinkConf().getReaderBatch();
        jdbcBatchSaver = new JdbcBatchSaver(jobContext);

        setName("JdbcSinkWorker-" +jobContext.getJobId());
    }

    @Override
    public void run() {
        while (!terminated) {
            synchronized (this) {
                while (paused) {
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt(); // Preserve interrupt status
                    }
                }
            }

            try {
                List<JdbcSinkEvent> sinkEvents = findBatch();
                new JdbcBatchSaver(jobContext).save(sinkEvents);

                //是否队列为空，没有数据可保存了？
                boolean sinkOver = RedisQueue.isEmpty(jobContext.getJobId());
                if (sinkOver) { //队列为空，工作线程休息一段时间
                    try {
                        Thread.sleep(SLEEP_TIME); // 模拟工作延迟
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt(); // Preserve interrupt status
                    }
                }
            } catch (Throwable e) {
                e.printStackTrace();
            }
        }
        log.info("Thread JdbcSinkWorker has been terminated.");
    }

    /**
     * 从redis队列中查找可以批量进行更新的数据变动事件。可批量更新的满足两个条件：
     * <ul>
     *     <li>同属于一张表</li>
     *     <li>且是同一种操作：INSERT、DELETE、UPDATE、INSERTWITHOUTKEY</li>
     * </ul>
     * 每个批次最大不超过预先设定的数值BATCH
     * @return
     */
    private List<JdbcSinkEvent> findBatch() {
        List<JdbcSinkEvent> sinkEvents = new ArrayList<>();
        String firstJson = RedisQueue.pop(jobContext.getJobId());

        if (null != firstJson) {
            log.debug(firstJson);
            try {
                JdbcSinkEvent firstSinkEvent = JdbcSinkEvent.buildFromJson(firstJson);
                sinkEvents.add(firstSinkEvent);
                for (int count = 1; count < BATCH; count++) {
                    String nextJson = RedisQueue.peek(jobContext.getJobId());
                    if (null == nextJson) {
                        break;
                    }

                    JdbcSinkEvent nextSinkEvent = JdbcSinkEvent.buildFromJson(nextJson);
                    if (isSimilar(firstSinkEvent, nextSinkEvent)) {
                        sinkEvents.add(nextSinkEvent);
                        RedisQueue.pop(jobContext.getJobId()); //记得出队
                    } else {
                        break;
                    }
                }
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        return sinkEvents;
    }

    private boolean isSimilar(JdbcSinkEvent first, JdbcSinkEvent next) {
        return first.getTable().equals(next.getTable())
                && first.getEventType() == next.getEventType();

    }

    /**
     * 恢复线程
     */
    public synchronized void resumeThread() {
        paused = false;
        notify();
    }

    public synchronized void pause() {
        paused = true;
    }

    /**
     * 终止线程
     */
    public void terminateThread() {
        terminated = true;
        // 如果线程正处于等待状态，唤醒线程使其退出
        if (paused) {
            resumeThread();
        }
    }
}
